package com.heima.user.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.model.user.pojos.ApConversationMessage;
import com.heima.user.service.ApConversationMessageService;
import com.heima.user.service.WorkflowService;
import lombok.extern.slf4j.Slf4j;
import okhttp3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Date;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class WorkflowServiceImpl implements WorkflowService {

    @Autowired
    private ApConversationMessageService conversationMessageService;

    @Value("${workflow.api.url:https://dify.aipfuture.com/v1/workflows/run}")
    private String workflowApiUrl;

    @Value("${workflow.api.key:your-api-key}")
    private String workflowApiKey;

    /**
     * 调用工作流并实时推送结果
     *
     * @param response HTTP响应对象
     * @param query    用户查询内容
     * @param userId   用户ID
     */
    @Override
    public void runWorkflow(HttpServletResponse response, String query, Integer userId) {
        // 设置SSE响应头
        response.setContentType("text/event-stream");
        response.setCharacterEncoding("utf-8");
        response.setHeader("Connection", "keep-alive");
        response.setHeader("Cache-Control", "no-cache");

        String msgId = UUID.randomUUID().toString();
        String workflowRunId = UUID.randomUUID().toString();
        String workflowId = UUID.randomUUID().toString();

        // 构建请求参数
        JSONObject requestBody = buildWorkflowRequest(query, userId, workflowRunId, workflowId);

        // 记录用户问题
        ApConversationMessage userMessage = new ApConversationMessage();
        userMessage.setMsgType(true); // 用户消息
        userMessage.setCreateTime(new Date());
        userMessage.setConversationId("11111111");
        userMessage.setMsgId(msgId);
        userMessage.setRequestId("");
        userMessage.setUserId(userId);
        userMessage.setContent(query);
        conversationMessageService.save(userMessage);

        // 发起工作流调用
        callWorkflowAndStreamResponse(response, requestBody, msgId, workflowRunId, userId);
    }

    /**
     * 构建工作流请求参数
     *
     * @param query         查询内容
     * @param userId        用户ID
     * @param workflowRunId 工作流运行ID
     * @param workflowId    工作流ID
     * @return 请求参数JSON对象
     */
    private JSONObject buildWorkflowRequest(String query, Integer userId, String workflowRunId, String workflowId) {
        JSONObject requestBody = new JSONObject();
        requestBody.put("response_mode", "streaming");
        requestBody.put("user", userId != null ? userId.toString() : "default-user");

        JSONObject inputs = new JSONObject();
        inputs.put("q", query);
        inputs.put("sys.user_id", userId != null ? userId.toString() : "default-user");
        inputs.put("sys.workflow_id", workflowId);
        inputs.put("sys.workflow_run_id", workflowRunId);

        requestBody.put("inputs", inputs);
        return requestBody;
    }

    /**
     * 调用工作流并流式返回结果
     *
     * @param response HTTP响应对象
     * @param request  请求参数
     * @param msgId    消息ID
     * @param userId   用户ID
     */
    private void callWorkflowAndStreamResponse(HttpServletResponse response, JSONObject request,
                                               String msgId, String workflowRunId, Integer userId) {
        try {
            OkHttpClient client = new OkHttpClient().newBuilder()
                    .connectTimeout(0, TimeUnit.MILLISECONDS)
                    .readTimeout(0, TimeUnit.MILLISECONDS)
                    .writeTimeout(0, TimeUnit.MILLISECONDS)
                    .build();

            MediaType mediaType = MediaType.parse("application/json");
            RequestBody body = RequestBody.create(mediaType, request.toJSONString());
            Request httpRequest = new Request.Builder()
                    .url(workflowApiUrl)
                    .method("POST", body)
                    .addHeader("Authorization", "Bearer " + workflowApiKey)
                    .addHeader("Content-Type", "application/json")
                    .build();

            CountDownLatch latch = new CountDownLatch(1);
            StringBuffer fullContent = new StringBuffer();

            client.newCall(httpRequest).enqueue(new Callback() {
                @Override
                public void onFailure(Call call, IOException e) {
                    log.error("工作流调用失败: {}", e.getMessage(), e);
                    try {
                        JSONObject errorMsg = new JSONObject();
                        errorMsg.put("event", "error");
                        errorMsg.put("message", "工作流调用失败");
                        sendSseEvent(response, errorMsg, true);
                    } catch (IOException ioException) {
                        log.error("发送错误消息失败: {}", ioException.getMessage());
                    } finally {
                        latch.countDown();
                    }
                }

                @Override
                public void onResponse(Call call, Response httpResponse) throws IOException {
                    try (ResponseBody responseBody = httpResponse.body()) {
                        if (!httpResponse.isSuccessful()) {
                            log.error("工作流调用失败，状态码: {}", httpResponse.code());
                            JSONObject errorMsg = new JSONObject();
                            errorMsg.put("event", "error");
                            errorMsg.put("message", "工作流调用失败，状态码: " + httpResponse.code());
                            sendSseEvent(response, errorMsg, true);
                            return;
                        }

                        // 处理流式响应
                        processStreamingResponse(response, responseBody, msgId, fullContent, userId);
                    } catch (Exception e) {
                        log.error("处理响应时发生错误: {}", e.getMessage(), e);
                    } finally {
                        // 保存完整回答到数据库
                        saveAssistantMessage(msgId, fullContent.toString(), userId);
                        latch.countDown();
                        try {
                            response.getWriter().close();
                        } catch (IOException e) {
                            log.error("关闭响应输出流失败: {}", e.getMessage());
                        }
                    }
                }
            });

            // 等待异步处理完成
            latch.await();
        } catch (Exception e) {
            log.error("执行工作流时发生异常: {}", e.getMessage(), e);
        }
    }

    /**
     * 处理流式响应
     *
     * @param response     HTTP响应对象
     * @param responseBody 响应体
     * @param msgId        消息ID
     * @param fullContent  完整内容缓冲区
     * @param userId       用户ID
     * @throws IOException IO异常
     */
    private void processStreamingResponse(HttpServletResponse response, ResponseBody responseBody,
                                          String msgId, StringBuffer fullContent, Integer userId) throws IOException {
        String line;
        while ((line = responseBody.source().readUtf8Line()) != null) {
            if (!StringUtils.hasText(line)) {
                continue;
            }

            try {
                // Dify的流式响应格式为 data: {...}
                if (line.startsWith("data:")) {
                    String jsonData = line.substring(5).trim();
                    processEventData(response, jsonData, fullContent);
                }
            } catch (Exception e) {
                log.warn("解析流式响应行时出错: {}，原始数据: {}", e.getMessage(), line);
                // 即使解析出错也继续处理后续数据
            }
        }

        // 发送结束事件
        JSONObject endMsg = new JSONObject();
        endMsg.put("event", "workflow_finished");
        endMsg.put("type", "end");
        sendSseEvent(response, endMsg, true);
    }

    /**
     * 处理单个事件数据
     * @param response HTTP响应对象
     * @param jsonData JSON数据
     * @param fullContent 完整内容缓冲区
     * @throws IOException IO异常
     */
    private void processEventData(HttpServletResponse response, String jsonData, StringBuffer fullContent) throws IOException {
        JSONObject eventObj = JSON.parseObject(jsonData);
        String event = eventObj.getString("event");

        if ("workflow_started".equals(event) || "node_started".equals(event)) {
            // 工作流状态事件，可选择性推送
            JSONObject statusMsg = new JSONObject();
            statusMsg.put("event", event);
            statusMsg.put("type", "status");
            sendSseEvent(response, statusMsg, false);
        } else if ("text_chunk".equals(event) && eventObj.containsKey("data")) {
            // 文本块事件（如果有的话）
            JSONObject dataObj = eventObj.getJSONObject("data");
            if (dataObj.containsKey("text")) {
                String text = dataObj.getString("text");
                if (org.apache.commons.lang3.StringUtils.isNotEmpty(text)) {
                    fullContent.append(text);
                    JSONObject contentMsg = new JSONObject();
                    contentMsg.put("event", "text_chunk");
                    contentMsg.put("type", "content");
                    contentMsg.put("content", text);
                    sendSseEvent(response, contentMsg, false);
                }
            }
        } else if (("node_finished".equals(event) || "workflow_finished".equals(event)) && eventObj.containsKey("data")) {
            // 节点完成或工作流完成事件，包含实际内容
            processFinishedEvent(response, eventObj, fullContent);
        }
    }

    /**
     * 处理完成事件（node_finished或workflow_finished）
     * @param response HTTP响应对象
     * @param eventObj 事件对象
     * @param fullContent 完整内容缓冲区
     * @throws IOException IO异常
     */
    private void processFinishedEvent(HttpServletResponse response, JSONObject eventObj, StringBuffer fullContent) throws IOException {
        String event = eventObj.getString("event");
        JSONObject dataObj = eventObj.getJSONObject("data");
        
        // 尝试从outputs.Group1提取内容
        if (dataObj.containsKey("outputs")) {
            JSONObject outputsObj = dataObj.getJSONObject("outputs");
            if (outputsObj.containsKey("Group1")) {
                String text = outputsObj.getString("Group1");
                if (org.apache.commons.lang3.StringUtils.isNotEmpty(text)) {
                    fullContent.append(text);
                    
                    JSONObject contentMsg = new JSONObject();
                    contentMsg.put("event", "content".equals(event) ? "text_chunk" : event);
                    contentMsg.put("type", "content");
                    contentMsg.put("content", text);
                    sendSseEvent(response, contentMsg, false);
                    return;
                }
            }
        }
        
        // 如果outputs.Group1没有内容，尝试从inputs.Group1提取（针对node_finished事件）
        if ("node_finished".equals(event) && dataObj.containsKey("inputs")) {
            JSONObject inputsObj = dataObj.getJSONObject("inputs");
            if (inputsObj.containsKey("Group1")) {
                String text = inputsObj.getString("Group1");
                if (org.apache.commons.lang3.StringUtils.isNotEmpty(text)) {
                    fullContent.append(text);
                    
                    JSONObject contentMsg = new JSONObject();
                    contentMsg.put("event", "text_chunk");
                    contentMsg.put("type", "content");
                    contentMsg.put("content", text);
                    sendSseEvent(response, contentMsg, false);
                }
            }
        }
    }

    /**
     * 发送SSE事件
     *
     * @param response HTTP响应对象
     * @param data     数据内容
     * @param flush    是否强制刷新
     * @throws IOException IO异常
     */
    private void sendSseEvent(HttpServletResponse response, JSONObject data, boolean flush) throws IOException {
        response.getWriter().write("data:" + data.toJSONString() + "\n\n");
        if (flush) {
            response.getWriter().flush();
        }
    }

    /**
     * 保存助手消息到数据库
     *
     * @param msgId   消息ID
     * @param content 消息内容
     * @param userId  用户ID
     */
    private void saveAssistantMessage(String msgId, String content, Integer userId) {
        try {
            ApConversationMessage assistantMessage = new ApConversationMessage();
            assistantMessage.setMsgType(false); // 助手消息
            assistantMessage.setCreateTime(new Date());
            assistantMessage.setConversationId("11111111");
            assistantMessage.setMsgId(msgId);
            assistantMessage.setRequestId("");
            assistantMessage.setUserId(userId);
            assistantMessage.setContent(content);
            conversationMessageService.save(assistantMessage);
        } catch (Exception e) {
            log.error("保存助手消息失败: {}", e.getMessage(), e);
        }
    }
}